package main;

import java.io.IOException;
import java.net.URI;
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import java.util.Map;

import net.sf.json.JSONArray;

import org.apache.commons.lang.StringUtils;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.filecache.DistributedCache;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.mapreduce.TableMapReduceUtil;
import org.apache.hadoop.hbase.mapreduce.TableOutputFormat;
import org.apache.hadoop.hbase.mapreduce.TableReducer;
import org.apache.hadoop.hbase.util.Bytes;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.input.KeyValueTextInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.util.GenericOptionsParser;


/**迁移机房，数据源改为hive表
 * 数据源Hive表
CREATE TABLE if not exists m_user_search_word(
	str_date string COMMENT '请勿使用此字段，仅作占位',
	platform string comment '平台',
	business_type string comment '业务线(zhaopin,fangchan,huangye...)',
	word string COMMENT '站内搜索词',
	disp_city_id int COMMENT '城市最小粒度id',
	disp_city1_name string COMMENT '展现一级城市',
	disp_city2_name string COMMENT '展现二级城市',
	disp_city3_name string COMMENT '展现三级城市',
	disp_city4_name string COMMENT '展现四级城市',
	disp_city5_name string COMMENT '展现五级城市',
	disp_cate_id int COMMENT '类别最小粒度id',
	disp_cate1_name string COMMENT '展现一级类别',
	disp_cate2_name string COMMENT '展现二级类别',
	disp_cate3_name string COMMENT '展现三级类别',
	disp_cate4_name string COMMENT '展现四级类别',
	disp_cate5_name string COMMENT '展现五级类别',
	search_count int COMMENT '搜索次数'	
* @author zhaoxiang
*/
@SuppressWarnings("deprecation")
public class MCityCateSearchWordTop50_V3 {
		//Mapper
		private static class MyMapper extends Mapper<LongWritable, Text, Text, IntWritable>
		{
			IntWritable counter = new IntWritable();
			public  Map<String,String> businessInfo=null;
			
			@Override
			protected void setup(Context context) throws IOException,InterruptedException 
			{
				try
				{
					Path[] paths = DistributedCache.getLocalCacheFiles(context.getConfiguration());
					businessInfo = CatchFileTools.getMapByCache(paths[0], "dict_disp_cate", "UTF-8", "\t", 2, 1);
				} 
				catch (Exception e) 
				{
					e.printStackTrace();
				}
			}
			
			public void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException 
			{
				String line=value.toString().replace("\\N", "");
				try 
				{
					JSONArray recod = JSONArray.fromObject(line.split("\t"));
					
					if(recod.size() < 17){
						return;
					}
					
					//日期，业务线，一级城市，二级类，搜索词，次数
					String stat_date = recod.getString(0);
					String platform = recod.getString(1);
					String word = recod.getString(3).replaceAll("_", "").trim();
					String cateId = recod.getString(10);
					String business_name = businessInfo.get(cateId);
					String disp_city1_name = recod.getString(5);
					String disp_cate1_name = recod.getString(11);
					String disp_cate2_name = recod.getString(12);
					int search_count = recod.getInt(16);
					counter.set(search_count);
					
					//搜索词既要统计在城市类别里，又要统计在上级汇总里面
					String [] cityCateArr1={disp_city1_name,business_name,disp_cate1_name,disp_cate2_name,stat_date,platform,word};
					String [] cityCateArr2={disp_city1_name,business_name,disp_cate1_name,"全部",stat_date,platform,word};
					String [] cityCateArr3={disp_city1_name,business_name,"全部","全部",stat_date,platform,word};
					String [] cityCateArr4={disp_city1_name,"全部","全部","全部",stat_date,platform,word};
					
					String [] cityCateArr5={"全部","全部","全部","全部",stat_date,platform,word};
					String [] cityCateArr6={"全部",business_name,"全部","全部",stat_date,platform,word};
					String [] cityCateArr7={"全部",business_name,disp_cate1_name,"全部",stat_date,platform,word};
					String [] cityCateArr8={"全部",business_name,disp_cate1_name,disp_cate2_name,stat_date,platform,word};
					
					context.write(new Text(StringUtils.join(cityCateArr1, "_")), counter);
					context.write(new Text(StringUtils.join(cityCateArr2, "_")), counter);
					context.write(new Text(StringUtils.join(cityCateArr3, "_")), counter);
					context.write(new Text(StringUtils.join(cityCateArr4, "_")), counter);
					context.write(new Text(StringUtils.join(cityCateArr5, "_")), counter);
					context.write(new Text(StringUtils.join(cityCateArr6, "_")), counter);
					context.write(new Text(StringUtils.join(cityCateArr7, "_")), counter);
					context.write(new Text(StringUtils.join(cityCateArr8, "_")), counter);
				} 
				catch (Exception e) 
				{
					e.printStackTrace();
				}
			}
		}
		
		//Reducer
		private static class MyReducer extends Reducer<Text, IntWritable, Text, IntWritable> 
		{
			private IntWritable counter = new IntWritable();
			@Override
			public void reduce(Text key, Iterable<IntWritable> values,Context context) throws IOException, InterruptedException 
			{	
				int searchCount=0;
				for (IntWritable v:values) 
				{
					try
					{
						searchCount = searchCount + v.get();
					}
					catch(Exception e)
					{
						e.printStackTrace();
						return;
					}
				}
				counter.set(searchCount);
				context.write(key, counter);
			}
		}
		
		
		//Mapreduce2  取top 50
		private static class MyMapper2 extends Mapper<Text, Text, Text, Text>
		{
			private Text k = new Text(); 
			private Text v = new Text(); 
			
			@Override
			public void map(Text key,Text value, Context context) throws IOException, InterruptedException 
			{
				String line=key.toString();
				try 
				{
					int index = line.lastIndexOf("_");
					
					if(index < 0)
						return;
					String rowKey = line.substring(0,index);
					String word = line.substring(index+1);
					k.set(rowKey);
					v.set(word+":"+value);
					context.write(k, v);
				} 
				catch (Exception e) 
				{
					e.printStackTrace();
				}
			}
		}
		
		//Reducer2
		private static class MyReducer2 extends TableReducer<Text, Text, NullWritable> 
		{
			class Pair implements Comparable<Pair>{
				private String word;
				private int count;
				public Pair(String word,int count){
					this.word=word;
					this.count=count;
				}
				public String getWord() {
					return word;
				}
				
				public int getCount() {
					return count;
				}
				
				@Override
				public int compareTo(Pair o) {
					return this.count -  o.count;
				}
				
			}
			
			public void add(List<Pair> topN,Pair p){
				int n = topN.size();
				if(n < 49){
					topN.add(p);
				}else if(n == 49){
					topN.add(p);
					Collections.sort(topN);
				}else{
					Pair minPair = topN.get(0);
					if(minPair.getCount() < p.getCount()){	//替换最小值
						topN.set(0, p);
						Collections.sort(topN);
					}
				}
			}
			
			 
			/**
			  * @param key: disp_city1_name , business_name, disp_cate1_name, disp_cate2_name, stat_date
			  * @param values [word1:count1, word2:count2]
			*/
			@Override
			public void reduce(Text key, Iterable<Text> values,Context context) throws IOException, InterruptedException 
			{
				List<Pair> topN = new ArrayList<Pair>(52);
				for(Text t : values){
						String line=t.toString();  // word:count
						int index = line.lastIndexOf(":");
						String word = line.substring(0,index);
						String count = line.substring(index+1);
						Pair pair = new Pair(word, Integer.parseInt(count));
						if(StringUtils.isBlank(pair.getWord()))
							continue;
						add(topN,pair);
				}
				//创建HBase中的RowKey
				byte[] rowKey=Bytes.toBytes(key.toString());
//				ImmutableBytesWritable rowKeyWritable=new ImmutableBytesWritable(rowKey);
				Put putrow = new Put(rowKey);
				int i = 0;
				for(Pair p : topN){
					//创建HBase中KeyValue对象,第一个参数为rowKey，第二个参数为列族，第三个参数为列，第四个参数为 具体的值
					byte[] family=Bytes.toBytes("cf");
					byte[] qualifier=Bytes.toBytes(p.getWord());
					byte[] hbaseValue=Bytes.toBytes(String.valueOf(p.getCount()));
					putrow.add(family, qualifier, hbaseValue);
					i++;
				}
				if(i == 0)
					return;
				context.write(NullWritable.get(), putrow);
			}
		}
		
		public static int getArgsParamsNum(String[] args){
			if(args==null || args.length==0){
				return 0;
			}else{
				return args.length;
			}
		}
		
		public static void main(String[] args) throws IOException, InterruptedException, ClassNotFoundException 
		{
			//job执行参数设置
			String JOB_NAME="CityCateSearchWordTop50";
			int REDUCE_TASK_NUM=10;
			
			String inPath = "/home/hdp_lbg_ecdata_dw/warehouse/hdp_lbg_ecdata_dw_defaultdb/d_search_word/stat_date=yyyy-MM-dd" ;
			String OPATH_1=String.format("/home/hdp_lbg_ecdata_dw/middata/zhaoxiang/yyyyMMdd",JOB_NAME);
//			String OPATH_2=String.format("/dsap/resultdata/zhaoxiang/SearchWordTopNJob/yyyyMMdd",JOB_NAME);
			
			List<String> dateList=null;
			// 字符串日期遍历
			Configuration configuration = new Configuration();
			 
			String[] otherArgs = null;
			try
			{
				otherArgs = new GenericOptionsParser(configuration, args).getRemainingArgs();
			} 
			catch (IOException e) 
			{
				e.printStackTrace();
				return;
			}
			
			int argsNum=getArgsParamsNum(otherArgs);
			
			String yestoday=DateUtil.getYestoday("yyyyMMdd");
			if(argsNum==0){
				dateList=DateUtil.getDayList(yestoday, yestoday);
			}else if(argsNum==1){
				dateList=DateUtil.getDayList(otherArgs[0], otherArgs[0]);
			}else if(argsNum==2){
				dateList=DateUtil.getDayList(otherArgs[0], otherArgs[1]);
			}
			
			 int result=0;
//			 dateList=DateUtil.getDayList("20150428", "20150505");
			for (String statDate : dateList) 
			{
				System.out.println(statDate);
				try
				{
					configuration.set("statDate", statDate);
					configuration.set("mapreduce.output.fileoutputformat.compress","false");
					Job job1 = new Job(configuration, JOB_NAME);
					job1.setJarByClass(MCityCateSearchWordTop50.class);
					job1.setNumReduceTasks(REDUCE_TASK_NUM);
					job1.setMapperClass(MyMapper.class);
					job1.setReducerClass(MyReducer.class);
					job1.setMapOutputKeyClass(Text.class);
					job1.setMapOutputValueClass(IntWritable.class);
					job1.setOutputKeyClass(Text.class);
					job1.setOutputValueClass(IntWritable.class);
					
					DistributedCache.addCacheFile(new URI("/home/hdp_lbg_ecdata_dw/warehouse/hdp_lbg_ecdata_dw_defaultdb/dict_disp_cate/000000_0#dict_disp_cate"), job1.getConfiguration());
					
					FileSystem hdfs = FileSystem.get(configuration);
					Path outPutPath = new Path(OPATH_1.replace("yyyyMMdd", statDate));
					if(hdfs.exists(outPutPath))
					{
						hdfs.delete(outPutPath, true);
					}
					String inDate = DateUtil.strToStr(statDate, "yyyyMMdd", "yyyy-MM-dd");
					System.out.println(inDate);
					FileInputFormat.addInputPath(job1,new Path(inPath.replace("yyyy-MM-dd", inDate)));
					FileOutputFormat.setOutputPath(job1,outPutPath); 
					System.out.println(job1.waitForCompletion(true) ? 0 : 1);
					
					 /**方法一，通过bulk——load写入到hbase中**/
					/**方法二，直接put写入到hbase中**/
					Job job2=new Job(configuration, JOB_NAME+"_2"); 
			        job2.setJarByClass(MCityCateSearchWordTop50_V3.class);
			        job2.setMapperClass(MyMapper2.class);
			        job2.setReducerClass(MyReducer2.class);
			        job2.setInputFormatClass(KeyValueTextInputFormat.class);
			        job2.setOutputFormatClass(TableOutputFormat.class);
			        
			        job2.setMapOutputKeyClass(Text.class);
			        job2.setMapOutputValueClass(Text.class);
			        
			        //以Job1的输出做为输入
			        FileInputFormat.addInputPath(job2, outPutPath);
			       
			        //初始化，设置输出格式，表名
			        TableMapReduceUtil.initTableReducerJob("SearchWordCount", MyReducer2.class, job2);  
			        result=job2.waitForCompletion(true)?0:1;
				} 
				catch (Exception e) 
				{
					e.printStackTrace();
					return;
				}
			}
	        
	        System.exit(result);
		}
		
	}
